package zy.learn.demo.structuredstreaming.source

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.sql.streaming.{OutputMode, Trigger}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, SparkSession}

object FileSource {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    val sparkConf = new SparkConf().set("spark.sql.shuffle.partitions", "3")

    val spark = SparkSession.builder()
      .master("local[2]")
      .config(sparkConf)
      .appName("FileSource")
      .getOrCreate()

    // 创建Schema
    val userSchema = StructType(StructField("name", StringType)::StructField("age", IntegerType)::StructField("sex", StringType)::Nil)
    // 以流模式加载csv文件
    val df: DataFrame = spark.readStream
      .format("csv")
      .schema(userSchema)
      .load("E:\\MyGit\\spark-project\\data\\ss")


    // 输出
    df.writeStream
      .format("console")
      .outputMode(OutputMode.Update())
      .trigger(Trigger.ProcessingTime(1000))
      .start()
      .awaitTermination()
  }
}
